﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Net;
using Protocol;
using System.Collections.Concurrent;
using System.Threading;
using System.Diagnostics.Contracts;
namespace MDS
{
    public abstract class QuotationManagerBase : IQuotationManager
    {
        protected QuotationManagerConfig _config;
        protected IQuotationProvider _quotationProvider;
        protected ITradeDaysProvider _tradeDaysProvider;
        protected IRepositoryFactory _repositoryFactory;

        protected IMessageRepository _messageRepository;
        protected ISnapshotRepository<Protocol.GWMKTBasicInfoANS> _basicSnapshotRepository;
        protected ISnapshotRepository<Protocol.GWMKTBidAskANS> _bidSnapshotRepository;
        protected ISnapshotRepository<Protocol.GWMKTTicksANS> _tickSnapshotRepository;
        protected HTSServer _htsServer;

        protected ConcurrentQueue<IMessage> _basicInfoBroadcastQueue = new ConcurrentQueue<IMessage>();
        protected ConcurrentQueue<IMessage> _msgBroadcastQueue = new ConcurrentQueue<IMessage>();
        protected Thread _basicInfoBroadcastThread;
        protected Thread _msgBroadcastThread;
        protected AutoResetEvent _newBasicInfoReceivedEvent = new AutoResetEvent(false);
        protected AutoResetEvent _newMsgReceivedEvent = new AutoResetEvent(false);

        public QuotationManagerBase(
            QuotationManagerConfig config,
            IQuotationProvider quotationProvider,
            ITradeDaysProvider tradeDaysProvider,
            IRepositoryFactory repositoryFactory)
        {

            Contract.Requires(config != null);
            Contract.Requires(quotationProvider != null);
            Contract.Requires(tradeDaysProvider != null);
            Contract.Requires(repositoryFactory != null);

            this._config = config;
            this._quotationProvider = quotationProvider;
            this._tradeDaysProvider = tradeDaysProvider;
            this._repositoryFactory = repositoryFactory;
            initRepositories();
            initHtsServer();
            initQuotationProviers();
            initBroadcastThreads();
        }

        private void initRepositories()
        {
            DateTime currentTime = DateTimeHelper.GetCurrentTime();
            DateTime currentTradeDay = this._tradeDaysProvider.GetTradeDay(currentTime);
            string key = RepositoryKeyGenerator.GenerateKey(this._config.Name, currentTradeDay);
            this._messageRepository = this._repositoryFactory.GetMessagePepository("Msg" + key);
            this._basicSnapshotRepository = this._repositoryFactory.GetSnapshotRepository<GWMKTBasicInfoANS>("Basic" + key);
            this._bidSnapshotRepository = this._repositoryFactory.GetSnapshotRepository<GWMKTBidAskANS>("Bid" + key);
            this._tickSnapshotRepository = this._repositoryFactory.GetSnapshotRepository<GWMKTTicksANS>("Tick" + key);
        }

        private void initBroadcastThreads()
        {
            ThreadStart basicTS = new ThreadStart(() => { broadCast(this._basicInfoBroadcastQueue, this._newBasicInfoReceivedEvent); });
            this._basicInfoBroadcastThread = new Thread(basicTS);
            this._basicInfoBroadcastThread.Start();
            ThreadStart msgTS = new ThreadStart(() => { broadCast(this._msgBroadcastQueue, this._newMsgReceivedEvent); });
            this._msgBroadcastThread = new Thread(msgTS);
            this._msgBroadcastThread.Start();
        }


        void initHtsServer()
        {
            this._htsServer = new HTSServer(this._config.Port, Log.LogHelper);
            this._htsServer.ConnectedEventHandler += new EventHandler(_htsServer_ConnectedEventHandler);
            this._htsServer.PackageReceivedEventHandler += new EventHandler(_htsServer_PackageReceivedEventHandler);
            this._htsServer.DisconnectedEventHandler += new EventHandler(_htsServer_DisconnectedEventHandler);
            this._htsServer.Start();
        }
        private void initQuotationProviers()
        {
            this._quotationProvider.MessagesReceived += new MultiMessageHandler<IMessage>(_QuotationProvider_OnMessagesReceived);
        }

        protected virtual void _QuotationProvider_OnMessagesReceived(IEnumerable<IMessage> messages)
        {
            if (messages == null)
                return;
            var basicInfos = from msg in messages
                             where msg is GWMKTBasicInfoANS
                             select msg as GWMKTBasicInfoANS;
            var bids = from msg in messages
                       where msg is GWMKTBidAskANS
                       select msg as GWMKTBidAskANS;
            var ticks = from msg in messages
                        where msg is GWMKTTicksANS
                        select msg as GWMKTTicksANS;

            enqueueBasicInfos(basicInfos);
            updateBasicInfoSnapshot(basicInfos);
            this._newBasicInfoReceivedEvent.Set();

            saveBidMsgs(bids);
            enqueueBids(bids);
            updateBidSnapshot(bids);


            saveTickMsgs(ticks);
            enqueueTicks(ticks);
            updateTickSnapshot(ticks);

            this._newMsgReceivedEvent.Set();

        }

        protected void enqueueBasicInfos(IEnumerable<GWMKTBasicInfoANS> basicInfos)
        {
            foreach (var bi in basicInfos)
                this._basicInfoBroadcastQueue.Enqueue(bi);
        }
        protected void enqueueBids(IEnumerable<GWMKTBidAskANS> bids)
        {
            foreach (var bid in bids)
                this._msgBroadcastQueue.Enqueue(bid);
        }
        protected void enqueueTicks(IEnumerable<GWMKTTicksANS> ticks)
        {
            foreach (var tick in ticks)
                this._msgBroadcastQueue.Enqueue(tick);
        }
        protected void saveBidMsgs(IEnumerable<GWMKTBidAskANS> bids)
        {

        }
        protected void saveTickMsgs(IEnumerable<GWMKTTicksANS> ticks)
        {

        }
        protected void updateBasicInfoSnapshot(IEnumerable<GWMKTBasicInfoANS> basicInfos)
        {
            if (basicInfos == null || basicInfos.Any() == false)
                return;
            this._basicSnapshotRepository.SaveSnapshot(basicInfos);
        }
        protected void updateBidSnapshot(IEnumerable<GWMKTBidAskANS> bids)
        {
            if (bids == null || bids.Any() == false)
                return;
            this._bidSnapshotRepository.SaveSnapshot(bids);
        }
        protected void updateTickSnapshot(IEnumerable<GWMKTTicksANS> ticks)
        {
            if (ticks == null || ticks.Any() == false)
                return;
            this._tickSnapshotRepository.SaveSnapshot(ticks);
        }



        protected void broadCast(ConcurrentQueue<IMessage> msgQueue, AutoResetEvent msgReceivedEvent)
        {
            while (true)
            {
                while (!msgQueue.IsEmpty)
                {
                    IMessage msg;
                    if (msgQueue.TryDequeue(out msg))
                    {
                        this._htsServer.BroadcastMessage(msg);
                    }
                }
                msgReceivedEvent.WaitOne();
            }
        }


        #region HTS事件处理
        void _htsServer_PackageReceivedEventHandler(object sender, EventArgs e)
        {
            PackageEventArgs ea = e as PackageEventArgs;
            if (ea != null && ea.RequestMessage != null && ea.RequestMessage is GWMKTReFillREQ && ea.CurrentConnection != null)
                onRefillRequest(ea.RequestMessage as GWMKTReFillREQ, ea.CurrentConnection as ServerConnection);
        }
        void _htsServer_ConnectedEventHandler(object sender, EventArgs e)
        {
            ConnectedEventArgs ea = e as ConnectedEventArgs;
            if (ea != null && ea.CurrentConnection != null)
            {
                ea.CurrentConnection.HeartbeatAvailable = false;
                onHTSServerConnected(ea.CurrentConnection as ServerConnection);
            }
        }
        void _htsServer_DisconnectedEventHandler(object sender, EventArgs e)
        {
            DisconnectedEventArgs ea = e as DisconnectedEventArgs;
            if (ea != null && ea.CurrentConnection != null)
            {
                onHTSServerDisconnected(ea.CurrentConnection as ServerConnection);
            }
        }
        #endregion

        #region 连线建立
        protected virtual void onHTSServerConnected(ServerConnection connection)
        {
            if (connection == null || connection.Available == false)
                return;
            logConnection(connection);
            trySendTradeTimeInfoMsg(connection);
            trySendBasicInfoSnapshot(connection);
            trySendBidSnapshot(connection);
            trySendMaxMessageSN(connection);
            connection.Broadcast = true;
        }
        protected static void logConnection(ServerConnection connection)
        {
            try
            {
                Log.Debug(string.Format("链接建立，IP={0}", connection.TcpClient.Client.RemoteEndPoint.ToString()));
            }
            catch { }
        }
        protected bool trySendTradeTimeInfoMsg(ServerConnection connection)
        {
            GWSYSSETTradeTimeANS tradeTimeInfo = new GWSYSSETTradeTimeANS();
            //tradeTimeInfo.LocalDatetime = TimeTool.Time2Int64(DateTime.Now);
            //tradeTimeInfo.TradeDate = TradeDayConf.GetInstance().GetTradeDate();
            //tradeTimeInfo.NextTradeDate = TradeDayConf.GetInstance().GetNextTradeDate();
            //tradeTimeInfo.TradeTimeZone = MarketConfManager.GetInstance().GetTradeSection(NetConfig.GetInstance().GetServerConf(CommServer.ServerIPEndPoint.Port).Name);
            bool succeed = connection.SendMessage(tradeTimeInfo);
            Log.Debug(string.Format("发送GWSYSSETTradeTimeANS，result={0}", succeed));
            return succeed;
        }
        protected bool trySendBasicInfoSnapshot(ServerConnection connection)
        {
            IEnumerable<GWMKTBasicInfoANS> basicInfos;
            bool succeed = this._basicSnapshotRepository.LoadSnapshot(out basicInfos);

            if (succeed && basicInfos != null)
            {
                foreach (var bid in basicInfos)
                {
                    if (connection.Available)
                    {
                        try
                        {
                            connection.SendMessage(bid);
                        }
                        catch (Exception ex)
                        {
                            succeed = false;
                            Log.Debug(ex);
                            break;
                        }

                    }
                    else
                    {
                        succeed = false;
                        break;
                    }

                }
            }
            return succeed;
        }
        protected bool trySendBidSnapshot(ServerConnection connection)
        {

            IEnumerable<GWMKTBidAskANS> bids;
            bool succeed = this._bidSnapshotRepository.LoadSnapshot(out bids);

            if (succeed && bids != null)
            {
                foreach (var bid in bids)
                {
                    if (connection.Available)
                    {
                        try
                        {
                            connection.SendMessage(bid);
                        }
                        catch (Exception ex)
                        {
                            succeed = false;
                            Log.Debug(ex);
                            break;
                        }

                    }
                    else
                    {
                        succeed = false;
                        break;
                    }
                }
            }


            return succeed;
        }
        protected bool trySendMaxMessageSN(ServerConnection connection)
        {
            GWMKTSETMaxMessSNANS maxMessSNREQ = new GWMKTSETMaxMessSNANS();
            maxMessSNREQ.TypeFlag = 1;
            maxMessSNREQ.MaxSN = getCurrentSN();
            bool succeed = connection.SendMessage(maxMessSNREQ);
            return succeed;
        }
        #endregion

        #region 连线断开
        protected virtual void onHTSServerDisconnected(ServerConnection connection)
        {
            logDisconnect(connection);
        }
        protected static void logDisconnect(ServerConnection connection)
        {
            try
            {
                Log.Debug(string.Format("链接断开，IP={0}", connection.TcpClient.Client.RemoteEndPoint.ToString()));
            }
            catch { }
        }
        #endregion

        #region 回补
        protected virtual void onRefillRequest(GWMKTReFillREQ refillRequest, ServerConnection connection)
        {
            Log.Debug(string.Format("收到回补请求，refillRequest.ActionFlag={0},refillRequest.BeginSN={1}, refillRequest.IsDirectlyBackFill={2}", refillRequest.ActionFlag, refillRequest.BeginSN, refillRequest.IsDirectlyBackFill));
            if (refillRequest.ActionFlag == '2')//从指定区间回补
            {
                int beginSN = refillRequest.BeginSN;
                try
                {
                    refillFromCertainSN(connection, beginSN);
                }
                catch (Exception ex)
                {
                    Log.Debug(ex);
                }
            }
            else if (refillRequest.ActionFlag == '4')//从最新序号回补
            {
                //不知道干啥
            }
            if (connection != null && connection.Available)
            {
                connection.Broadcast = true;
            }
        }

        private void refillFromCertainSN(ServerConnection connection, int beginSN)
        {
            if (beginSN <= 1)
            {
                beginSN = 1;
            }
            int currentSN = this.getCurrentSN();
            for (int sn = beginSN; sn <= currentSN; sn++)
            {
                IMessage msg = this.getMessage(sn);
                if (connection.Available)
                {
                    if (msg != null)
                        connection.SendMessage(msg);
                }
                else
                    return;
            }

        }
        #endregion

        protected int getCurrentSN()
        {
            return this._messageRepository.GetCurrentSN();
        }

        protected IMessage getMessage(int sequenceNo)
        {
            return this._messageRepository.GetMessage(sequenceNo);
        }

        protected Protocol.GWMKTTicksANS getTickInSnapshot(string stkId)
        {
            return _tickSnapshotRepository.GetMessageByStkId(stkId);
        }


        public void Start()
        {
            this._quotationProvider.Start();
        }

        public void Stop()
        {
            this._quotationProvider.Stop();
        }
    }
}
